package liuzhou.spark.project

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  *
  */
object kfk {

  def main(args: Array[String]): Unit = {
    val  ssc = new StreamingContext("local[*]","liuzhou",Seconds(5))

    //消费者参数
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "node1:9092,node2:9092,node3:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "aqi_1",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    val topics = Array("flumeTopic")

    val dstream = KafkaUtils.createDirectStream[String,String](
      ssc ,
      PreferConsistent,
      Subscribe[String,String](topics,kafkaParams)
    )

    dstream.map(_.value()).print()


    ssc.sparkContext.setLogLevel("WARN")

    ssc.start()
    ssc.awaitTermination()
  }
}
